package cn.openread.demo.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;

public class CanalTools {

    /**
     * 如果采用rocketMQ去进行流转消息，仅需要监控topic即可
     * <p>
     * {
     * "data": [
     * {
     * "member_id": "2",
     * "account": "simon",
     * "password": "E9BkXt9ZEx6GL9Nx5yqBLGPtQt-sRyBD3J7OF1aKtJEiXJS_zacs_OTuCVyYWDC9YdCYcr7j_uODGrof_6ImHe2QffRO6J5YML2220RamkzVDKMAq0RkV9DMvNf6RxleSrY=$18$0dd953422227ab816cf2a0597da726edcada0a00eefb94a66fe774a9$e566e09495bf1d6436c43f2c530ed3f918634ebcee38872d8d62d3c760938ad9",
     * "auth_method": "local",
     * "description": "1234",
     * "email": "812135023@qq.com",
     * "phone": "15038302175",
     * "avatar": "/uploads/201803/avatar_151e20f8806a6d37_small.jpeg",
     * "role": "1",
     * "status": "0",
     * "create_time": "2017-09-21 23:26:14",
     * "create_at": "0",
     * "last_login_time": "2019-03-09 12:32:45",
     * "real_name": ""
     * }
     * ],
     * "database": "online_markdown_doc_db",
     * "es": 1552140176000,
     * "id": 2,
     * "isDdl": false,
     * "mysqlType": {
     * "member_id": "int(11)",
     * "account": "varchar(100)",
     * "password": "varchar(1000)",
     * "auth_method": "varchar(50)",
     * "description": "varchar(2000)",
     * "email": "varchar(100)",
     * "phone": "varchar(255)",
     * "avatar": "varchar(1000)",
     * "role": "int(11)",
     * "status": "int(11)",
     * "create_time": "datetime",
     * "create_at": "int(11)",
     * "last_login_time": "datetime",
     * "real_name": "varchar(255)"
     * },
     * "old": [
     * {
     * "description": "123"
     * }
     * ],
     * "sql": "",
     * "sqlType": {
     * "member_id": 4,
     * "account": 12,
     * "password": 12,
     * "auth_method": 12,
     * "description": 12,
     * "email": 12,
     * "phone": 12,
     * "avatar": 12,
     * "role": 4,
     * "status": 4,
     * "create_time": 93,
     * "create_at": 4,
     * "last_login_time": 93,
     * "real_name": 12
     * },
     * "table": "md_members",
     * "ts": 1552140176321,
     * "type": "UPDATE"
     * }
     */
    public static void main(String args[]) {
        // 创建链接
        CanalConnector connector = CanalConnectors
                .newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            int totalEmtryCount = 1200;
            while (emptyCount < totalEmtryCount) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n",
                    // batchId, size);
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

            System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            EventType eventType = rowChage.getEventType();
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));

            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                } else {
                    System.out.println("-------> before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------> after");
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }
}
